package com.net.xpay.sync.client;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringEscapeUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.List;


@Slf4j
@Component
public abstract class AbstractSyncEventHandler implements SyncEventHandler {

    @Autowired
    @Qualifier("commonDataSource")
    private DataSource commonDataSource;

    @Override
    @SuppressWarnings("unchecked")
    public void handleUpdateEvent(CanalSyncContext context) {
        log.debug(getClass().getName() + "|handleUpdateEvent: " + context.getAfterBeanJson());
        Preconditions.checkArgument(context.getId() != null);

        String updateSql = getUpdateSql(context);
        log.info("handleUpdateEvent updateSql={}", updateSql);

        Connection connection = null;
        Statement statement = null;
        try {
            connection = commonDataSource.getConnection();
            statement = connection.createStatement();
            statement.execute(updateSql);
            log.debug(getClass().getName() + "|handleUpdateEvent success update {}", context.getAfterBeanJson());
        } catch (SQLException e) {
            log.error("handleUpdateEvent error|beforeJson={}|afterJson={}|tableName={}", context.getBeforeBeanJson(), context.getAfterBeanJson(), context.getTableName(), e);
            throw new RuntimeException("happen SQLException");
        } finally {
            if (statement != null) {
                try {
                    statement.close();
                } catch (SQLException e) {
                    log.error("close statement error", e);
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (SQLException e) {
                    log.error("close connection error", e);
                }
            }
        }

    }

    @Override
    @SuppressWarnings("unchecked")
    public void handleInsertEvent(CanalSyncContext context) {
        log.debug(getClass().getName() + "|handleInsertEvent: " + context.getAfterBeanJson());
        Preconditions.checkArgument(context.getId() != null);

        String insertSql = getInsertSql(context);
        log.debug("handleInsertEvent insertSql={}", insertSql);

        Connection connection = null;
        Statement statement = null;
        try {
            connection = commonDataSource.getConnection();
            statement = connection.createStatement();
            statement.execute(insertSql);

            log.debug(getClass().getName() + "|handleInsertEvent success insert {}", context.getAfterBeanJson());
        } catch (SQLException e) {
            //主键冲突的 重复插入 不用管
            if (e instanceof com.mysql.jdbc.exceptions.jdbc4.MySQLIntegrityConstraintViolationException
                    || e instanceof com.mysql.jdbc.exceptions.MySQLIntegrityConstraintViolationException) {
                log.info("handleInsertEvent MySQLIntegrityConstraintViolationException", e);
                return;
            }
            log.error("handleInsertEvent error|beforeJson={}|afterJson={}|tableName={}", context.getBeforeBeanJson(), context.getAfterBeanJson(), context.getTableName(), e);
            throw new RuntimeException("happen SQLException");
        } finally {
            if (statement != null) {
                try {
                    statement.close();
                } catch (SQLException e) {
                    log.error("close statement error", e);
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (SQLException e) {
                    log.error("close connection error", e);
                }
            }
        }

    }

    @Override
    public void handleDeleteEvent(CanalSyncContext context) {
        //删除语句不处理
        log.info(getClass().getName() + "handleDeleteEvent: " + context.getBeforeBeanJson());
        Preconditions.checkArgument(context.getId() != null);

        String sql = "delete from " + getTargetTableName(context.getTableName()) + " where id = " + context.getId();

        //查询是否有老数据
        Connection connection = null;
        Statement statement = null;
        try {
            connection = commonDataSource.getConnection();
            statement = connection.createStatement();
            statement.execute(sql);
            log.info(getClass().getName() + "|handleDeleteEvent success delete table={}|id={}", context.getTableName(), context.getId());
        } catch (SQLException e) {
            log.error("handleDeleteEvent error|beforeJson={}|afterJson={}|tableName={}", context.getBeforeBeanJson(), context.getAfterBeanJson(), context.getTableName(), e);
            throw new RuntimeException("happen SQLException");
        } finally {
            if (statement != null) {
                try {
                    statement.close();
                } catch (SQLException e) {
                    log.info("close statement error", e);
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (SQLException e) {
                    log.info("close connection error", e);
                }
            }
        }
    }


    private String getInsertSql(CanalSyncContext context) {
        CanalEntry.RowData rowData = context.getRowData();
        List<CanalEntry.Column> columns = rowData.getAfterColumnsList();

        //字段
        StringBuilder fields = new StringBuilder();
        //值
        StringBuilder values = new StringBuilder();
        columns.forEach((column) -> {
            //丢弃空值
            if (!column.getIsNull()) {
                fields.append("`").append(column.getName()).append("`").append(",");
                values.append("'").append(StringEscapeUtils.escapeSql(column.getValue()).replace("\\", "\\\\")).append("'").append(",");
            }
        });

        String fieldsStr = fields.substring(0, fields.length() - 1);
        String valuesStr = values.substring(0, values.length() - 1);


        return "INSERT INTO `" + getTargetTableName(context.getTableName()) + "` (" + fieldsStr + ") VALUES (" + valuesStr + ")";
    }

    private String getUpdateSql(CanalSyncContext context) {
        CanalEntry.RowData rowData = context.getRowData();
        List<CanalEntry.Column> columns = rowData.getAfterColumnsList();

        StringBuilder sql = new StringBuilder("update " + getTargetTableName(context.getTableName()) + " set ");
        columns.forEach((column) -> {
            if (!column.getIsNull()) {
                sql.append("`").append(column.getName()).append("`").append("=")
                        .append("'").append(StringEscapeUtils.escapeSql(column.getValue()).replace("\\", "\\\\")).append("',");
            } else {
                sql.append("`").append(column.getName()).append("`").append("=").append("NULL,");
            }
        });
        return sql.substring(0, sql.length() - 1) + " where id=" + context.getId();
    }

    private String getTargetTableName(String tableName) {
        if (Strings.isNullOrEmpty(tablePrefix())) {
            return tableName;
        }
        return tablePrefix() + tableName;
    }
}

